package com.gjy.design.observer.disruptor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-11-08 09:16:51
 */
public class HighPerformancePubSubSystem {

    private final TopicManager topicManager;
    private final RingBuffer<DisruptorEvent> ringBuffer;
    private final SubscriberEventHandler[] eventHandlers;
    private final AtomicBoolean running = new AtomicBoolean(false);

    public HighPerformancePubSubSystem(int bufferSize, int handlerCount) {
        this.topicManager = new TopicManager();

        // 创建Disruptor
        this.eventHandlers = new SubscriberEventHandler[handlerCount];
        Disruptor<DisruptorEvent> disruptor = createDisruptor(bufferSize, handlerCount);
        this.ringBuffer = disruptor.getRingBuffer();
        disruptor.start();
        // 启动系统
        start();
    }

    private Disruptor<DisruptorEvent> createDisruptor(int bufferSize, int handlerCount) {
        DisruptorEventFactory factory = new DisruptorEventFactory();

        // 创建Disruptor（使用多生产者模式）
        Disruptor<DisruptorEvent> disruptor = new Disruptor<>(
                factory,
                bufferSize,
                Executors.defaultThreadFactory(),
                ProducerType.MULTI,
                new BlockingWaitStrategy()
        );

        // 创建事件处理器
        SubscriberEventHandler[] handlers = new SubscriberEventHandler[handlerCount];
        for (int i = 0; i < handlerCount; i++) {
            handlers[i] = new SubscriberEventHandler("Handler-" + i);
            this.eventHandlers[i] = handlers[i];
        }

        // 设置处理器（广播模式，所有处理器都会处理每个事件）
        disruptor.handleEventsWith(handlers);

        return disruptor;
    }

    public void start() {
        if (running.compareAndSet(false, true)) {
            // Disruptor在创建后会自动启动
        }
    }

    public void shutdown() {
        if (running.compareAndSet(true, false)) {
            // 关闭事件处理器
            for (SubscriberEventHandler handler : eventHandlers) {
                if (handler != null) {
                    handler.shutdown();
                }
            }
        }
    }

    // 订阅主题
    public void subscribe(String topic, Subscriber subscriber) {
        topicManager.subscribe(topic, subscriber);
    }

    // 取消订阅
    public void unsubscribe(String topic, Subscriber subscriber) {
        topicManager.unsubscribe(topic, subscriber);
    }

    // 发布消息
    public boolean publish(String topic, Object message) {
        return publish(topic, message, generateMessageId());
    }

    public boolean publish(String topic, Object message, String messageId) {
        if (!topicManager.hasSubscribers(topic)) {
            return false; // 没有订阅者，直接返回
        }

        CopyOnWriteArraySet<Subscriber> subscribers = topicManager.getSubscribers(topic);
        if (subscribers == null || subscribers.isEmpty()) {
            return false;
        }

        // 发布到Disruptor
        long sequence = ringBuffer.next();
        try {
            DisruptorEvent event = ringBuffer.get(sequence);
            event.set(new MessageEvent(topic, message, messageId), subscribers);
        } finally {
            ringBuffer.publish(sequence);
        }

        return true;
    }

    private String generateMessageId() {
        return java.util.UUID.randomUUID().toString();
    }

    // 获取系统状态
    public SystemStatus getStatus() {
        return new SystemStatus(
                running.get(),
                ringBuffer.getBufferSize(),
                ringBuffer.remainingCapacity(),
                topicManager.topicSubscribers.size()
        );
    }

}
